package spring.cloud.tasks.executor.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.ExecutorServiceUtil;
import spring.cloud.tasks.common.ThreadFactory;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.curator.CuratorFactory;
import spring.cloud.tasks.common.curator.CuratorInfo;
import spring.cloud.tasks.executor.ConnectionReConnectListener;
import spring.cloud.tasks.executor.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.executor.task.execute_statistics.ExecuteStatisticsReportTaskService;
import spring.cloud.tasks.executor.executor.task.execute_statistics.ExecuteStatisticsResetTaskService;
import spring.cloud.tasks.executor.executor.waitingExecuteTasks.SlidingTimeTaskBucketQueueService;

import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.locks.ReentrantLock;

import static spring.cloud.tasks.common.Threads.sleep;

/**
 * <pre>
 * /tasks/taskId/taskInstanceId/taskRoundId/taskItemIndex 实例创建,进行续期。
 * /taskItemAssigns/executor/TaskItem list 任务分配列表
 * /executors/executes/executor/taskId/taskInstanceId/taskRoundId/taskItemIndex 执行信息备份。两个小时内预分配未执行+已经执行且超过超时时间后一个小时的任务。如果执行器重启，那么这部分的任务也应该保存下来。
 * /executors/registers/executorId
 * </pre>
 */
@Slf4j
@Setter
@Getter
public class Executor {
    //
    public static volatile boolean needStopFlag = false;
    public TaskIdToTaskExecutorMap taskIdToTaskExecutorMap = new TaskIdToTaskExecutorMap();
    private String executorId;
    //
    private String namespaceKey;
    private Curator curator;
    private ConnectionReConnectListener namespaceDataManagerConnectionReConnectListener;
    //
    private ExecutorInitAndTaskInitService executorInitAndTaskInitService;
    private TaskMap taskMap = new TaskMap();//
    private SlidingTimeTaskBucketQueueService slidingTimeTaskBucketQueueService;
    //
    private ScheduledThreadPoolExecutor taskItemExecuteTimeoutTaskScheduledThreadPoolExecutor;
    private ExecuteStatisticsResetTaskService executeStatisticsResetTaskService;
    private ExecuteStatisticsReportTaskService executeStatisticsReportTaskService;
    //
    private ReentrantLock lock = new ReentrantLock();
    private Thread restartThread;
    private volatile boolean needRestartExecutor = false;
    //
    private volatile boolean isShutdown;

    //
    public Executor(String namespaceKey, String executorId) {
        this.namespaceKey = namespaceKey;
        this.executorId = executorId;
        initRestartExecutorThread();
    }

    public static boolean setNeedStopFlag(boolean needStopFlag) {
        return Executor.needStopFlag = needStopFlag;
    }

    public static boolean getNeedStopFlag() {
        return Executor.needStopFlag;
    }

    /**
     * 执行器启动，该方法给外部调用的
     */
    public void start() throws Exception {
        lock.lockInterruptibly();
        try {
            try {
                start0();
            } catch (Throwable t) {
                log.info("执行器启动失败:" + executorId + ",执行器组件资源清理中...,稍后重启执行器组件资源");
                stop0(false);
                throw t;
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 执行器停止，该方法给外部调用的
     */
    public void stop() throws Exception {
        stop(true);
    }

    private void initRestartExecutorThread() {
        final String threadName = "executor:" + executorId + "-restart-thread";
        this.restartThread = new Thread(new Runnable() {
            @Override
            public void run() {
                restartExecutorThread();
            }
        });
        this.restartThread.setName(threadName);
        this.restartThread.setDaemon(false);
        this.restartThread.start();
    }

    private void restartExecutorThread() {
        try {

            while (true) {
                if (isShutdown) {
                    return;
                }
                //通过needRestartExecutor变量控制执行器重启,该标志位置为true表示需要重启执行器。
                if (needRestartExecutor) {
                    try {
                        needRestartExecutor = false;
                        restartExecutor();
                    } catch (InterruptedException e) {
                        throw e;
                    } catch (Throwable t) {
                        needRestartExecutor = true;
                        log.warn("重启线程出现异常,稍后再进行重启:" + t.getMessage(), t);
                    }
                }
                sleep();//优先满足执行器运行状态正常再进行休眠
            }
        } catch (InterruptedException e) {
            log.warn("执行器重启线程检测到InterruptedException,执行器重启线程终止");
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 清理之前的执行器组件资源,同时创建新的执行器组件资源。
     */
    public void restartExecutor() throws Exception {
        lock.lockInterruptibly();
        try {
            if (isShutdown) {
                return;
            }
            stop0(false);
            try {
                start0();
            } catch (Throwable t) {
                log.info("执行器启动失败:" + executorId + ",组件资源清理中...,稍后重启执行器组件资源清");
                stop0(false);
                throw t;
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 执行器初始化
     */
    private void start0() throws Exception {
        log.info("连接zk,命名空间:" + namespaceKey);
        String addressList = "127.0.0.1:2181";
        String digest = "cxl:cxl";
        CuratorInfo curatorInfo = new CuratorInfo();
        curatorInfo.setAddressList(addressList);
        curatorInfo.setDigest(digest);
        curatorInfo.setNamespaceKey(namespaceKey);
        //该代码用来测试 zk连接超时的情况。 namespaceConfiguration.setTestSessionLostArgs();
        initNamespaceDataManager(curatorInfo);
        log.info("注册Executor到zk,命名空间:" + namespaceKey);
        registerExecutor();
        //
        slidingTimeTaskBucketQueueService = new SlidingTimeTaskBucketQueueService();
        //
        log.info("启动任务统计信息的线程");
        executeStatisticsReportTaskService = new ExecuteStatisticsReportTaskService(executorId);
        executeStatisticsReportTaskService.start();
        log.info("启动任务统计信息重置的线程");
        executeStatisticsResetTaskService = new ExecuteStatisticsResetTaskService(executorId);
        executeStatisticsResetTaskService.start();
        //
        log.info("创建任务超时处理线程池");
        taskItemExecuteTimeoutTaskScheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(Math.max(2, Runtime.getRuntime().availableProcessors() / 2), new ThreadFactory(executorId + "-timeout-watchdog", false));
        taskItemExecuteTimeoutTaskScheduledThreadPoolExecutor.setRemoveOnCancelPolicy(true);
        //
        log.info("注册任务执行器创建监听器");
        executorInitAndTaskInitService.registerTaskWatcher(this);
        //
        needStopFlag = false;//全局可见性。防止底层控制标志不可见。
        log.info("执行器组件资源启动成功:" + executorId);
    }

    private void initNamespaceDataManager(CuratorInfo curatorInfo) throws Exception {
        try {
            curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_false);
            long sessionId = curator.getSessionId();
            namespaceDataManagerConnectionReConnectListener = new ConnectionReConnectListener(executorId, sessionId) {
                @Override
                public void restartExecutor0() {
                    needRestartExecutor = true;
                }
            };
            curator.addConnectionStateListener(namespaceDataManagerConnectionReConnectListener);
        } catch (Exception e) {
            throw e;
        }
    }

    private void registerExecutor() throws Exception {
        try {
            log.info("注册执行器到ZooKeeper开始");
            executorInitAndTaskInitService = new ExecutorInitAndTaskInitService(executorId, curator);
            executorInitAndTaskInitService.registerExecutor();
            log.info("注册执行器到ZooKeeper结束");
        } catch (Exception e) {
            log.error("注册执行器到ZooKeeper异常", e);
            throw e;
        }
    }


    /**
     * Executor关闭
     *
     * @param shutdownGracefully Executor优雅退出： 把自己从集群中拿掉，现有的任务不停； 一直到全部任务都执行完毕，再真正退出； 设置一定超时时间，如果超过这个时间仍未退出，则强行中止
     */
    public void stop(boolean shutdownGracefully) throws Exception {
        if (isShutdown) {
            return;
        }
        lock.lockInterruptibly();
        try {
            if (isShutdown) {
                return;
            }
            stop0(shutdownGracefully);
            restartThread.interrupt();
            isShutdown = true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 经过测试zk节点挂了,调用该方法能够正确的关闭执行器相关的组件资源。add by huihui。
     * 能够处理的原因是，关闭相关的资源都进行异常catch处理。
     * 需要提到的一个资源关闭是执行器ip节点是瞬时节点。
     * 断开网络后瞬时节点会自动消失(如果断开网络后可能会在会话有效期内存在，那么在启动的时候还会检查到该节点存在，那么需要等待节点自己消失再进行注册成功)。
     */
    private void stop0(boolean shutdownGracefully) throws Exception {
        lock.lockInterruptibly();
        needStopFlag = true;//全局可见性。防止底层控制标志不可见。
        try {
            log.info("执行器" + executorId + "开始停止");
            //不接受新的任务以及把自己从zk反注册
            if (executorInitAndTaskInitService != null) {
                executorInitAndTaskInitService.unRegisterTaskWatcher();
                executorInitAndTaskInitService.unRegisterExecutor();
            }
            //

            //
            // 先关闭统计信息上报，因为上报的zk结点为servers/xxx/xxx，
            // 如果放在下线后再关闭，则导致ExecutorCleanService执行后，
            // 仍然在上报统计信息，垃圾结点由此而生
            if (executeStatisticsReportTaskService != null) {
                executeStatisticsReportTaskService.stop();
            }
            if (executeStatisticsResetTaskService != null) {
                executeStatisticsResetTaskService.stop();
            }
            if (taskItemExecuteTimeoutTaskScheduledThreadPoolExecutor != null) {
                taskItemExecuteTimeoutTaskScheduledThreadPoolExecutor.shutdown();
            }
            if (shutdownGracefully) {
                tryCompletedTask();
            }
            shutdownTaskExecutors();
            if (slidingTimeTaskBucketQueueService != null) {
                slidingTimeTaskBucketQueueService.stop();//必须等执行上面操作后再进行清空。但是需要避免本执行器的任务因为执行器反复异常导致反复执行。
                slidingTimeTaskBucketQueueService = null;
            }
            taskMap.taskIdToTaskItemCollectionMap.clear();
            //可能这期间存在和zk交互的场景,最后关闭没有影响
            if (namespaceDataManagerConnectionReConnectListener != null) {
                namespaceDataManagerConnectionReConnectListener.close();
            }
            if (curator != null) {
                curator.removeConnectionStateListener(namespaceDataManagerConnectionReConnectListener);
                curator.close();
            }
            log.info("执行器" + executorId + "停止成功");
        } finally {
            lock.unlock();
        }
    }


    private void tryCompletedTask() {
        Map<String, TaskExecutor> taskIdToTaskExecutorMap = this.taskIdToTaskExecutorMap.getTaskIdToTaskExecutorMap();
        Set<Entry<String, TaskExecutor>> entrySet = taskIdToTaskExecutorMap.entrySet();
        if (entrySet == null || entrySet.size() == 0) {
            return;
        }
        //executor优雅退出的默认超时时间(单位：精确到秒，默认1分钟)。
        long shutdownTimeout = 60 * 1000L;
        long startTime = System.currentTimeMillis();
        do {
            sleep();
            //
            boolean hasRunning = false;
            for (Entry<String, TaskExecutor> entry : entrySet) {
                TaskExecutor taskExecutor = entry.getValue();
                if (taskExecutor != null) {
                    if (taskExecutor.getTaskExecuteEngine().isRunning()) {
                        hasRunning = true;
                        break;
                    }
                }
            }
            if (!hasRunning) {
                log.info("所有任务停止");
                break;
            }
        } while (System.currentTimeMillis() - startTime < shutdownTimeout);

    }

    private void shutdownTaskExecutors() {
        Map<String, TaskExecutor> taskIdToTaskExecutorMap = this.taskIdToTaskExecutorMap.removeAndGetTaskIdToTaskExecutorMap();
        if (taskIdToTaskExecutorMap == null) {
            return;
        }
        Iterator<String> iterator = taskIdToTaskExecutorMap.keySet().iterator();
        ExecutorService executorService = Executors.newCachedThreadPool(new ThreadFactory(executorId + "-shutdownTaskExecutors-thread", true));
        List<Future<?>> futureList = new ArrayList<>();
        while (iterator.hasNext()) {
            final String taskId = iterator.next();
            final TaskExecutor taskExecutor = taskIdToTaskExecutorMap.get(taskId);
            if (taskExecutor != null) {
                taskExecutor.needStop();
                futureList.add(executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            taskExecutor.shutdown();
                        } catch (Throwable t) {
                            log.error("关闭任务执行器失败:" + t.getMessage(), t);
                        }
                    }
                }));
            }
        }
        for (Future<?> future : futureList) {
            try {
                future.get();
            } catch (Exception e) {
                log.error("wait 'taskExecutor.shutdown()' 失败:" + e.getMessage(), e);
            }
        }
        ExecutorServiceUtil.shutdownNow(executorService);
        log.info("关闭任务执行器完毕");
    }


}
